// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved.
// See License.txt in the project root for license information.
using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Runtime.Serialization.Formatters.Binary;
using System.Text;
using System.Threading;
using System.Threading.Actors;
using System.Threading.Tasks;

namespace WordCount
{
    class MapActor : FabricAdHocActorBase, IDisposable
    {
        private int m_numReducers;
        private int m_index;
        private FabricActorClient m_client;


        public MapActor(int index, int numReducers)
            : base("mapper" + index)
        {
            m_numReducers = numReducers;
            m_index = index;
        }

        public string Name { get { return "mapper" + m_index; } }
        public string FullName { get { return "fabric:/actor/adhoc/" + Name; } }

        public void Initialize(FabricActorClient client, string assemblyName, byte[] assemblyBytes)
        {
            m_client = client;
            client.AddAssembly(assemblyName, assemblyBytes);
            client.CallMethod("InitializeMapper", new object[] { m_index, m_numReducers });
        }

        public Task MapAsync(string filename)
        {
            return m_client.CallMethodAsyncVoid("MapFunction", new object[] { filename });
        }

        public void MarkComplete()
        {
            m_client.CallMethod("DoneMapping", new object[0]);
        }

        [ActorMethod]
        public static object InitializeMapper(dynamic state, dynamic[] args)
        {
            state.index = (int)args[0];
            state.numReducers = (int)args[1];
            return null;
        }

        // These delimit words in a text file
        public static readonly char[] s_delimiters = new char[] { ' ', '\t', '\n', '\r', '.', ';', ':', ',', '!', '?' };

        // These are words in which we are not interested.
        public static readonly HashSet<string> s_uninterestingWords = new HashSet<string>(
            new string[] { "a", "all", "an", "and", "are", "as", "be", "by", "but", "do", "for", "had", 
                           "has", "have", "he", "her", "hers", "him", "his", "i", "if", "in", "is", "it", "me", 
                           "my", "not", "of", "on", "or", "shall", "she", "so", "that", "the", "thee", "them", "then", 
                           "they", "thine", "this", "thou", "to", "thy", "was", "we", "what", "which", "will", "with", 
                           "you", "your" });
        
        // Processes a text file
        // Limitation: This isn't going to work in Azure, since it reads files from the client's local disk.
        // This assumes your client and 5 node development cluster are all on the same machine.
        // We could fix this by sending the file's contents to this method as a parameter.
        [ActorMethod]
        public static object MapFunction(dynamic state, dynamic[] args)
        {
            var numReducers = (int) state.numReducers;
            string filename = args[0];
            string justTheName = Path.GetFileNameWithoutExtension(filename);

            state.Publish("Update", justTheName, (int)state.index);
            // Initialize our counts for each reducer
            Dictionary<string, int>[] counts = new Dictionary<string, int>[numReducers];
            for (int i = 0; i < numReducers; i++) counts[i] = new Dictionary<string, int>();

            int totalWordCount = 0;
            using (var fs = new FileStream(filename, FileMode.Open, FileAccess.Read))
            using (var sr = new StreamReader(fs))
            {

                // Perform our local count, into segregated dictionaries
                string line;
                int lineCount = 0;
                
                while ((line = sr.ReadLine()) != null)
                {
                    line = line.ToLower();
                    var words = line.Split(s_delimiters, StringSplitOptions.RemoveEmptyEntries);
                    foreach (var word in words)
                    {
                        totalWordCount++;
                        if (s_uninterestingWords.Contains(word)) continue;
                        int reducerIndex = word[0] % numReducers;
                        var reducerDictionary = counts[reducerIndex];
                        if (reducerDictionary.ContainsKey(word)) reducerDictionary[word] += 1;
                        else reducerDictionary[word] = 1;

                        if ((totalWordCount % 1000) == 0)
                            state.Publish("Update", String.Format("{0}\n{1}", justTheName, totalWordCount), (int)state.index);
                    }
                    
                    // Artificial delay
                    lineCount++;
                    if ((lineCount % 20) == 0)
                    {
                        int elapsedTime = Environment.TickCount;
                        int targetTime = elapsedTime + 2;
                        int newTime = elapsedTime;
                        while (newTime < targetTime)
                        {
                            newTime = Environment.TickCount;
                        }
                    }
                }
            }

            // Output the counts to the proper reducers
            BinaryFormatter bf = new BinaryFormatter();
            for (int i = 0; i < numReducers; i++)
            {
                if (counts[i].Count > 0)
                {
                    string outdir = Environment.GetFolderPath(Environment.SpecialFolder.MyDocuments) + "\\MRFiles";
                    if (!Directory.Exists(outdir)) Directory.CreateDirectory(outdir);

                    //var outname = String.Format("{0}-from-{1}-to-{2}", Guid.NewGuid().ToString(), state.index, i);
                    var outname = String.Format("{0}-from-{1}-to-{2}", justTheName, state.index, i);
                    //var outpath = String.Format("{0}\\{1}", Environment.SpecialFolder.MyDocuments + "\\MRFiles\\", outname);  //generate path to settings.dat
                    var outpath = String.Format("{0}\\{1}", outdir, outname);  //generate path to settings.dat
                    using (var stream = new FileStream(outpath, FileMode.Create, FileAccess.Write))
                    {
                        bf.Serialize(stream, counts[i]);
                        stream.Flush();
                    }
                    var proxy = state.GetActorProxy("fabric:/actor/adhoc/reducer" + i);
                    proxy.Reduce(outpath);
                }
                
            }

            state.Publish("Update", "<idle>", (int)state.index);
            return null;
        }

        // Called externally when we are done processing all of our text files.
        [ActorMethod]
        public static object DoneMapping(dynamic state, dynamic[] args)
        {
            var numReducers = (int)state.numReducers;
            var myIndex = (int)state.index;
            for (int i = 0; i < numReducers; i++)
            {
                var proxy = state.GetActorProxy("fabric:/actor/adhoc/reducer" + i);
                proxy.Request("MappingComplete", new object[] { myIndex });
            }
            state.Publish("Update", "Done", (int)state.index);

            return null;
        }



        public void Dispose()
        {
            if (m_client != null)
            {
                m_client.Dispose();
                m_client = null;
            }
        }
    }
}
